package sbc.group1.lab1.mozartspaces;

import java.io.Serializable;
import java.net.URI;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.mozartspaces.capi3.CountNotMetException;
import org.mozartspaces.capi3.Selector;
import org.mozartspaces.core.ContainerReference;
import org.mozartspaces.core.Entry;
import org.mozartspaces.core.MzsCoreException;
import org.mozartspaces.core.MzsCoreRuntimeException;
import org.mozartspaces.core.RequestFuture;
import org.mozartspaces.core.TransactionReference;
import org.mozartspaces.core.MzsConstants.RequestTimeout;
import org.mozartspaces.core.util.Nothing;
import org.mozartspaces.notifications.Notification;
import org.mozartspaces.notifications.NotificationListener;
import org.mozartspaces.notifications.NotificationManager;
import org.mozartspaces.notifications.Operation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import sbc.group1.lab1.common.Car;

abstract class AbstractTester implements Runnable, NotificationListener{
    
    public static final Logger log = LoggerFactory.getLogger(AbstractTester.class);
    
    private final Object notificationLock = new Object();
    private static final ArrayList<Selector> selectors = new ArrayList<Selector>();
    private volatile boolean isWaitingForNotification = false;
    protected final SpaceClient space;
    private volatile boolean running;
    private TransactionReference transaction;
    private NotificationManager notificationManager;
    private Notification notificationNotTested;
    private Notification notificationPartiallyTested;
    private final ExecutorService executor = Executors.newFixedThreadPool(10);
    
    static {
        selectors.add(CarCoordinator.newSelector());
    }
    
    public AbstractTester(URL configFile, URI spaceURI) throws MzsCoreException {
        space= new SpaceClient();
        space.init(configFile, spaceURI);
        notificationManager = new NotificationManager(space.getCore());
    }
    
    
    @Override
    public void run() {
        log.info("Starting");
        running=true;
        try {
            createNotifications();
            while(running) {
                try {
                    doWork();
                    log.debug("Car test done.");
                } catch (MzsCoreException e1) {
                    log.error("Car test failed!",e1);
                } catch (InterruptedException e) {
                    log.info("Interrupted! Stopping...");
                    running=false;
                    return;
                }catch (MzsCoreRuntimeException e) {
                    if (e.getCause() instanceof InterruptedException) {
                        log.info("Interrupted! Stopping...");
                    }else {
                        log.error("", e);
                    }
                    running=false;
                    return;
                }
                transaction = null;
            }
        } catch (MzsCoreException e) {
            log.error("",e);
            return;
        } catch (InterruptedException e) {
           log.error("",e);
           return;
        }finally {
        	try {
        		if (transaction!=null) {
        			space.getCapi().rollbackTransaction(transaction);
        		}
			} catch (Exception e) {
			}
			try {
                destroyNotifications();
            } catch (Exception e) {
            }
            executor.shutdownNow();
        	space.getCore().shutdown(true);
        }
    }
    
    private void doWork() throws MzsCoreException, InterruptedException {
        transaction = space.getCapi().createTransaction(
                RequestTimeout.INFINITE, space.getSpaceUri());
        try {
            synchronized (this) {
                if (!testTestedFirst()) {
                    waitForNotification();
                }
            }
        }catch(InterruptedException e) {
            space.getCapi().rollbackTransaction(transaction);
            throw e;
        }catch(MzsCoreException e) {
            space.getCapi().rollbackTransaction(transaction);
            throw e;
        }
    }
    
    private boolean testTested() throws MzsCoreException {
        Car car;
        try {
            car = space.getCapi().<Car>take(getPartiallyTestedInContainer(), 
                    selectors, RequestTimeout.TRY_ONCE, transaction).get(0);
        }catch (CountNotMetException e) {
            return false;
        }
        markAndWrite(car);
        return true;
    }
    
    private boolean testUntested() throws MzsCoreException {
        Car car;
        try {
            car = space.getCapi().<Car>take(space.getCompletedCarContainer(), 
                    selectors, RequestTimeout.TRY_ONCE, transaction).get(0);
        }catch (CountNotMetException e) {
            return false;
        }
        markAndWrite(car);
        return true;
    }
    
    private boolean testTestedFirst() throws MzsCoreException {
        Car car;
        try {
            car = space.getCapi().<Car>take(getPartiallyTestedInContainer(), 
                    selectors, RequestTimeout.TRY_ONCE, transaction).get(0);
        }catch (CountNotMetException e) {
            try {
                car= space.getCapi().<Car>take(space.getCompletedCarContainer(), 
                        selectors, RequestTimeout.TRY_ONCE, transaction).get(0);
            } catch (CountNotMetException e1) {
                return false;
            }
        }
        markAndWrite(car);
        return true;
    }
    
    protected abstract void markCar(Car car);
    
    private void markAndWrite(Car car) throws MzsCoreException {
        markCar(car);
        ContainerReference writeContainer;
        if (car.isCompletelyTested()) {
            writeContainer=space.getTestedCarContainer();
        }else {
            writeContainer=getPartiallyTestedOutContainer();
        }
        
        RequestFuture<Nothing> writeOp = 
            space.getAcapi().write(writeContainer, RequestTimeout.INFINITE, 
                transaction, new Entry(car));
        executor.execute(new FailedTransactionHandler(
                writeOp,
                space.getAcapi().commitTransaction(transaction),
                transaction, space.getCapi()));
        transaction=null;
    }
    
    protected abstract ContainerReference getPartiallyTestedOutContainer();
    
    protected abstract ContainerReference getPartiallyTestedInContainer();

    private void waitForNotification() throws InterruptedException {
        isWaitingForNotification=true;
        while(isWaitingForNotification()) {
            wait();
        }
    }
    @Override
    public void entryOperationFinished(
            Notification source,
            Operation operation, List<? extends Serializable> entries) {
        synchronized (notificationLock) {
            synchronized (this) {
            if (isWaitingForNotification()) {
                try {
                    boolean success;
                    if (source == notificationPartiallyTested) {
                       success = testTested();
                    }else if(source == notificationNotTested) {
                        success = testUntested();
                    }else {
                        log.warn("Received obsolete notification.");
                        return;
                    }
                    if (success) {
                        isWaitingForNotification=false;
                        notifyAll();
                    }
                }catch (MzsCoreException e) {
                    isWaitingForNotification=false;
                    try {
                        space.getCapi().rollbackTransaction(transaction);
                    } catch (MzsCoreException e1) {
                    }
                    notifyAll();
                    log.error("",e);
                }
            }
            }
        }
    }
    
    private boolean isWaitingForNotification() {
        return isWaitingForNotification;
    }
    
    
    private void destroyNotifications() throws MzsCoreException {
        notificationPartiallyTested.destroy();
        notificationNotTested.destroy();
    }


    private void createNotifications() throws MzsCoreException, InterruptedException {
        HashSet<Operation> writeOp = new HashSet<Operation>();
        writeOp.add(Operation.WRITE);
        notificationPartiallyTested=notificationManager.createNotification(
                getPartiallyTestedInContainer(), 
                this, writeOp, transaction, null);
        notificationNotTested=notificationManager.createNotification(
                space.getCompletedCarContainer(), 
                this, writeOp, transaction, null);
    }
}
